package com.study.kafka;

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.admin.ListTopicsResult;
import org.apache.kafka.common.KafkaFuture;

import java.util.Properties;
import java.util.Set;


/**
 * @author liuguibin
 * @date 2021-01-15 10:03
 * 获取所有topic
 */
public class kafkatest {
    public static String gettopics() {
        AdminClient kafkaConn = null;
        try {
            JSONArray jsonArray = new JSONArray();

            Properties prop = new Properties();
            prop.put("bootstrap.servers", "188.188.23.161:9092");
            prop.put("retries", 0);
            prop.put("request.timeout.ms", 1000);

            kafkaConn = AdminClient.create(prop);
            ListTopicsResult listTopicsResult = kafkaConn.listTopics();
            KafkaFuture<Set<String>> topics = listTopicsResult.names();

            /*Set<String> strings =   new HashSet<>();
            strings.add("1");
            strings.add("2");*/
            topics.get().forEach((k) -> {
                JSONObject jsonObject = new JSONObject();
                jsonObject.put("name", k);
                jsonArray.add(jsonObject);
            });
            return jsonArray.toString();
        } catch (Exception e) {
            e.printStackTrace();
            return "topic查询错误";
        } finally {
            kafkaConn.close();
        }

    }
}
